package org.senseframework.support.orm.hbase.datasource;

import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.senseframework.support.orm.hbase.config.HBaseTableScanner;
import org.senseframework.support.orm.hbase.serializer.ISerialize;
import org.senseframework.support.orm.hbase.serializer.SerializeType;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.data.hadoop.hbase.HbaseTemplate;

import java.util.concurrent.ConcurrentHashMap;

/**
 * @author wangqian
 * @date 18/4/14 上午11:55
 */

public class HBaseDataSource implements ApplicationContextAware, BeanFactoryPostProcessor {

    private HbaseTemplate hbaseTemplate;

    private static ConcurrentHashMap<Class, HbaseTemplate> hbaseTemplateMap = new ConcurrentHashMap<Class, HbaseTemplate>();

    public HbaseTemplate getHbaseTemplate() {
        return hbaseTemplate;
    }

    public void setHbaseTemplate(HbaseTemplate hbaseTemplate) {
        this.hbaseTemplate = hbaseTemplate;
    }

    private static ISerialize defaultSerialize;

    public static ISerialize getDefaultSerialize() {
        return defaultSerialize;
    }

    public void setSerializer(SerializeType serializer) {
        try {
            HBaseDataSource.defaultSerialize = serializer.serialize.newInstance();
        } catch (Exception e) {
            throw new RuntimeException("init " + serializer.name() + " serializer fail");
        }
    }

    private boolean autoCreateTable;
    private String[] annotationPackages;
    private static ApplicationContext applicationContext = null;
    private ApplicationContext context = null;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        HBaseDataSource.applicationContext = applicationContext;
    }

    public static HbaseTemplate getSource(Class clazz) {
        if (hbaseTemplateMap.containsKey(clazz)) {
            return hbaseTemplateMap.get(clazz);
        } else {
            if (applicationContext != null) {
                HbaseTemplate template = applicationContext.getBean(HbaseTemplate.class);
                hbaseTemplateMap.put(clazz, template);
                return template;
            } else {
                throw new RuntimeException("applicationContext must not null !");
            }
        }
    }

    public boolean isAutoCreateTable() {
        return autoCreateTable;
    }

    public void setAutoCreateTable(boolean autoCreateTable) {
        this.autoCreateTable = autoCreateTable;
    }

    public String[] getAnnotationPackages() {
        return annotationPackages;
    }

    public void setAnnotationPackages(String[] annotationPackages) {
        this.annotationPackages = annotationPackages;
    }

    @Override
    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
        HBaseTableScanner scanner = new HBaseTableScanner((BeanDefinitionRegistry) beanFactory);
        scanner.setResourceLoader(this.context);
        scanner.registerDefaultFilters();
        scanner.scan(annotationPackages);
        if (autoCreateTable) {
            try {
                HBaseAdmin admin = new HBaseAdmin(hbaseTemplate.getConfiguration());
                new AutoCreateTable(admin).start(applicationContext);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
